package belf.migrate.api.taskconf;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.model.ProgressModel;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.table.schema.CatalogTable;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.api.util.FieldCase;
import belf.migrate.api.util.QuoteType;
import org.apache.commons.lang3.StringUtils;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/** This class is used to store the context of the job. e.g. the table schema, catalog...etc. */
public final class JobContext implements Serializable {

    private static final long serialVersionUID = -1L;

    // 缓存源端的表元数据
    // 缓存DatabaseIdentifier.database(schema).tableName -> tableSchema。KEY统一为大写。
    private final Map<String, CatalogTable> sourcetableMap =
            new ConcurrentHashMap<>(16);
    // 缓存目标端的表元数据
    private final Map<String, CatalogTable> sinktableMap =
            new ConcurrentHashMap<>(16);

    /** job对应的任务配置 */
    private final TaskConf taskConf;

    /** 本次job对应的id，用于追踪任务执行情况 */
    private final String jobId;

    /** 源端的{@link AbstractJdbcCatalog}对象 */
    private AbstractJdbcCatalog sourceCatalog;

    /** 目标端的{@link AbstractJdbcCatalog}对象 */
    private AbstractSinkCatalog sinkCatalog;

    /** 模块进度 */
    private Map<String, Object> progress = new HashMap<>(16);

    public JobContext(String jobId, TaskConf taskConf) {
        this.jobId = jobId;
        this.taskConf = taskConf;
    }

    public String getJobId() {
        return this.jobId;
    }

    public TaskConf getTaskConf() {
        return taskConf;
    }

    public AbstractJdbcCatalog getSourceCatalog() {
        return sourceCatalog;
    }

    public void setSourceCatalog(AbstractJdbcCatalog sourceCatalog) {
        this.sourceCatalog = sourceCatalog;
    }

    public AbstractSinkCatalog getSinkCatalog() {
        return sinkCatalog;
    }

    public void setSinkCatalog(AbstractSinkCatalog sinkCatalog) {
        this.sinkCatalog = sinkCatalog;
    }

    /**
     * Put table schema.注意tableMap的主键是大写的
     *
     * @param catalog 数据库类型，参考{@link DatabaseIdentifier}
     * @param fullTableName fulll table name, include database name or schema name
     * @param catalogTable Catalog Table
     */
    public void addSourceTable(String catalog, String fullTableName, CatalogTable catalogTable) {
        String key = (catalog + "-" + fullTableName).toUpperCase();
        sourcetableMap.put(key, catalogTable);
    }

    /**
     * Get source table schema.
     *
     * @param fullTableName fulll table name, include database name or schema name
     * @return table schema.
     */
    public CatalogTable getSourceTable(String catalog, String fullTableName) {
        String key = (catalog + "-" + fullTableName).toUpperCase();
        return sourcetableMap.get(key);
    }

    public Map<String, CatalogTable> getSourcetableMap() {
        return sourcetableMap;
    }

    /**
     * Put sink table schema.注意tableMap的主键是大写的
     *
     * @param catalog 数据库类型，参考{@link DatabaseIdentifier}
     * @param fullTableName fulll table name, include database name or schema name
     * @param catalogTable Catalog Table
     */
    public void addSinkTable(String catalog, String fullTableName, CatalogTable catalogTable) {
        String key = (catalog + "-" + fullTableName).toUpperCase();
        sinktableMap.put(key, catalogTable);
    }

    /**
     * Get sink table schema.
     *
     * @param fullTableName fulll table name, include database name or schema name
     * @return table schema.
     */
    public CatalogTable getSinkTable(String catalog, String fullTableName) {
        String key = (catalog + "-" + fullTableName).toUpperCase();
        return sinktableMap.get(key);
    }

    public Map<String, CatalogTable> getSinktableMap() {
        return sinktableMap;
    }

    /**
     * 对源端，从数据库名和schema名中二选一，若schema非空优先选择schema
     * @return
     */
    public String getValidSourceDatabase() {
        if (StringUtils.isNotBlank(taskConf.getSourceConf().getSchemaName())) {
            return taskConf.getSourceConf().getSchemaName();
        }
        return taskConf.getSourceConf().getDatabaseName();
    }

    public QualifiersModel getSourceQualifiersModel() {
        FieldCase fieldCase = FieldCase.getFieldCase(getTaskConf().getSourceConf().getFieldCase());
        DatabaseType databaseType = DatabaseType.getDatabaseType(getTaskConf().getSourceConf().getCatalogName());
        boolean isQuoted = getTaskConf().getSourceConf().isQuote();
        String leftQuote = isQuoted? databaseType.getQuoteType().getValue() : QuoteType.NONE.getValue();
        String rightQuote = leftQuote;
        if (leftQuote.length() == 2) {
            rightQuote = leftQuote.substring(1);
            leftQuote = leftQuote.substring(0, 1);
        }
        return new QualifiersModel(leftQuote, rightQuote, fieldCase, databaseType.getQualifiedType());
    }

    public QualifiersModel getSinkQualifiersModel() {
        FieldCase sinkFieldCase = FieldCase.getFieldCase(getTaskConf().getSinkConf().getFieldCase());
        DatabaseType databaseType = DatabaseType.getDatabaseType(getTaskConf().getSinkConf().getCatalogName());
        boolean isQuoted = getTaskConf().getSinkConf().isQuote();
        String leftQuote = isQuoted? databaseType.getQuoteType().getValue() : QuoteType.NONE.getValue();
        String rightQuote = leftQuote;
        if (leftQuote.length() == 2) {
            rightQuote = leftQuote.substring(1);
            leftQuote = leftQuote.substring(0, 1);
        }
        return new QualifiersModel(leftQuote, rightQuote, sinkFieldCase, databaseType.getQualifiedType());
    }
    /**
     * 对目标端，从数据库名和schema名中二选一，若schema非空优先选择schema
     * @return
     */
    public String getValidSinkDatabase() {
        if (StringUtils.isNotBlank(taskConf.getSinkConf().getSchemaName())) {
            return taskConf.getSinkConf().getSchemaName();
        }
        return taskConf.getSinkConf().getDatabaseName();
    }

    /**
     * 返回一张表的全名，优先是schema.table，再次是database.table
     * @param tableName
     * @return
     */
    public String getFullTableName(String tableName) {
        String prefixTable = getValidSinkDatabase();
        String simpleTableName = tableName;
        if (tableName.contains(".")) {
            simpleTableName = StringUtils.substringAfterLast(tableName, ".");
        }
        return prefixTable + "." + simpleTableName;
    }

    public void release() {
        try {
            getSourceCatalog().close(); //释放源端数据库连接
        } catch (Exception e){
        }

        try {
            getSinkCatalog().close(); //释放目标端数据库连接
        } catch (Exception e) {
        }
    }

    /**
     * 添加进度条
     * @param key
     * @param total
     */
    public void addProgress(String key, int total) {
        ProgressModel progressModel;
        if (progress.containsKey(key) && progress.get(key) instanceof ProgressModel) {
            progressModel = (ProgressModel) progress.get(key);
            progressModel.setCurrent(progressModel.getCurrent() + 1);
            progressModel.setTotal(total);
        } else {
            progressModel = new ProgressModel(1, total);
        }
        progress.put(key, progressModel);
    }

    public void setTotalProgress(int total) {
        progress.put("total", total);
    }

    public Map<String, Object> getProgress() {
        return progress;
    }
}
